package com.me.kafka.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * kafka测试消费者
 *
 * @author Wuzhiming
 */
@Component
@Slf4j
public class TestConsumer {

    @KafkaListener(topics = {"test"})
    public void testConsumerMessage(ConsumerRecord<String, String> record) {
        log.info("简单——接收到消息：" + record.topic() + "/" + record.key() + "/" + record.partition() + "/" + record.value());
    }

      //2.并发消费
    @KafkaListener(groupId = "kafka-topic",topics = "kafka-topic-003",containerFactory = "batchFactory",errorHandler = "myConsumerAwareErrorHandler")
    public void consurmTopic003(List<ConsumerRecord<String,String>> consumerRecords, Consumer consumer, Acknowledgment acknowledgment){
        log.info("consumerRecords size is:{}",consumerRecords.size());
        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            log.info("【当前线程thredName is】:{},【offset is】:{},【msg is】:{},【topic is】:{},【patition is】:{}",Thread.currentThread().getName(),
                    consumerRecord.offset(),consumerRecord.value(),
                    consumerRecord.topic(),consumerRecord.partition());
        }
        //手动异步提交
        acknowledgment.acknowledge();
    }

}
